package com.chukun.flink.table.api.sql;

import com.chukun.flink.table.bean.ClickBean;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author chukun
 * @version 1.0.0
 * @description 时间窗口join基本操作
 *  以一个小时的滚动窗口，统计每个窗口下每个用户访问的url数量，以及每个用户点击数据流中最小id和该id对应的url地址
 * @createTime 2022年06月03日 10:27:00
 */
public class TimeWindowJoinTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<ClickBean> clickStream = env.fromCollection(PrepareData.getClicksData());

        SingleOutputStreamOperator<ClickBean> timedDataStream = clickStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx) -> new WatermarkGenerator<ClickBean>() {
                    @Override
                    public void onEvent(ClickBean clickBean, long l, WatermarkOutput watermarkOutput) {}

                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        watermarkOutput.emitWatermark(new Watermark(System.currentTimeMillis()));
                    }
                }).withTimestampAssigner((clickBean, timestamp) -> clickBean.getTime().getTime())
        );

        Table inputTable = tableEnv.fromDataStream(timedDataStream, $("id"), $("user"), $("VisitTime").rowtime(), $("url"));

        tableEnv.createTemporaryView("Clicks", inputTable);

        // 以一个小时的滚动窗口，统计每个窗口下每个用户访问的url数量，以及每个用户点击数据流中最小id和该id对应的url地址
        String sql = "select temp.name, temp.minId, temp.number, url, temp.between_start, temp.between_time from (" +
                "select user as name, " +
                "count(url) as number, " +
                "min(id) as minId, " +
                "tumble_rowtime(VisitTime, interval '1' hour ) as between_time," + // 包含滚动窗口的上限时间戳
                "tumble_start(VisitTime, interval '1' hour ) as between_start " + // 包含滚动窗口的下限时间戳
                "from Clicks " +
                "group by tumble(VisitTime, interval '1' hour), user" +
                ") temp left join Clicks on temp.minId = Clicks.id " +
                "and Clicks.VisitTime <=temp.between_time and  Clicks.VisitTime >= temp.between_time - interval '1' hour";

        Table result = tableEnv.sqlQuery(sql);

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(result);

        rowDataStream.print("window-join");

        env.execute("TimeWindowJoinTemplate");

    }
}
